package com.edu.realtime.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import com.edu.realtime.util.MyKafkaUtil;

/**
 * Created on 2022/10/19.
 *
 * @author Topus
 * @desc 考试域：试卷问题信息 DwdTestPaperQuestion
 */
public class DwdTestPaperQuestion {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        // 检查点
        // env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        // env.getCheckpointConfig().setCheckpointTimeout(60000L);
        // env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30), Time.seconds(3)));
        // env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        // env.getCheckpointConfig()
        // .enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // env.setStateBackend(new HashMapStateBackend());
        // env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/edu/ck");
        // System.setProperty("HADOOP_USER_NAME", "root");

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_test_paper_question_group"));

        Table resultTb = tableEnv.sqlQuery("select data['id'] id, data['paper_id'] paper_id,\n"
            + "data['question_id'] question_id, data['score'] score, data['publisher_id'] publisher_id, ts\n"
            + "from topic_db\n" + "where `table`='test_paper_question' and `type`='insert'");

        tableEnv.createTemporaryView("result_table", resultTb);

        tableEnv.executeSql("create table dwd_test_paper_question(\n" + "id string,\n" + "paper_id string,\n"
            + "question_id string,\n" + "score string,\n" + "publisher_id string,\n" + "ts string,\n"
            + "primary key(id) not enforced\n" + ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_test_paper_question"));

        tableEnv.executeSql("insert into dwd_test_paper_question select * from result_table");
    }
}
